package com.zengqingfa.rocketmq.consumer;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 *
 * @fileName: MqDemo01Listener
 * @author: zengqf3
 * @date: 2021-4-2 22:59
 * @description:
 */
@Component
@Slf4j
public class MqDemo01Listener implements MessageListenerConcurrently {

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        log.info("receive  msgs:{}", JSON.toJSONString(msgs));
        if (CollectionUtils.isEmpty(msgs)) {
            log.info("receive blank msgs");
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        MessageExt messageExt = msgs.get(0);
        String msg = new String(messageExt.getBody());

        //模拟消费
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            log.error("失败：{}", e.getMessage());
        }
        log.info("receive msg: {}.", msg);

        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
